package com.kingsoft.dc.khaos.module.spark.util

import com.kingsoft.dc.khaos.module.spark.request.encode.EncoderFactory
import com.kingsoft.dc.khaos.util.Logging
import org.apache.spark.sql.Row
import org.apache.spark.sql.types.StructType
import org.elasticsearch.hadoop.serialization.field.FieldExtractor
import org.elasticsearch.spark.serialization.ScalaMapFieldExtractor

import scala.collection.mutable.ArrayBuffer

class UserDefinedDataFrameFieldExtractor extends ScalaMapFieldExtractor with Logging {
  val encoderFactory = new EncoderFactory()

  def encode(value: String): String = {
    encoderFactory.getEncoderUtil("md5").encode(value)
  }

  protected def parse(target: AnyRef, str: String): ArrayBuffer[String] = {
    var parseStr = str
    var template = ArrayBuffer[String]()
    while (parseStr.contains("{")) {
      val startPattern: Int = parseStr.indexOf("{")
      template += parseStr.substring(0, startPattern)
      val endPattern: Int = parseStr.indexOf("}")
      val nestedString: String = parseStr.substring(startPattern + 1, endPattern)
      val value = getFiledValue(target, nestedString.toLowerCase)

      if (value == null) {
        template += ""
        //        log.error(s"the field ${nestedString} value is null")
      } else {
        template += value.toString
      }
      parseStr = parseStr.substring(endPattern + 1).trim
    }
    template += parseStr
    return template
  }

  //generate doc id by the value of fields
  def getDocId(target: AnyRef, fields: String): String = {
    var id = ""
    val fieldValues = parse(target, fields)
    val idValue = fieldValues.toArray.mkString("")
    id = encode(idValue)
    id
  }

  def getFiledValue(target: AnyRef, field: String): AnyRef = {
    var obj = target
    obj = obj match {
      case (row: Row, struct: StructType) => {
        //        val index = struct.fieldNames.indexOf(field)
        val index = FieldNameUtils.indexOfIgnoreCase(struct.fieldNames, field)
        if (index < 0) {
          FieldExtractor.NOT_FOUND
        } else {
          row(index).asInstanceOf[AnyRef]
        }
      }
    }
    return obj
  }

  override protected def extractField(target: AnyRef): AnyRef = {
    var obj = target
    for (in <- 0 until getFieldNames.size()) {
      val field = getFieldNames.get(in)
      //the field is config to generate the doc id
      if (field.contains("{") && field.contains("}")) {
        getDocId(target, field)
      }
      obj = obj match {
        case (row: Row, struct: StructType) => {
          // 返回加密的值作为doc_id
          Row(getDocId(target, field)).get(0).asInstanceOf[AnyRef]
          //          val index = struct.fieldNames.indexOf(field)
          //          if (index < 0) {
          //            FieldExtractor.NOT_FOUND
          //          } else {
          //            println("==>" + row(index).toString)
          //            val row1: Row = Row(getDocId(target, field))
          //            Row(getDocId(target, field)).get(0).asInstanceOf[AnyRef]
          //          }
        }
        case _ => {
          super.extractField(target)
        }
      }
    }
    return obj
  }
}